package com.axdoc.walletbiz.record.service.impl;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import javax.ws.rs.Consumes;
import javax.ws.rs.GET;
import javax.ws.rs.POST;
import javax.ws.rs.Path;
import javax.ws.rs.Produces;
import javax.ws.rs.QueryParam;
import javax.ws.rs.core.MediaType;

import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import com.alibaba.fastjson.JSONObject;
import com.axdoc.walletbiz.record.async.DataAsync;
import com.axdoc.walletbiz.record.dao.IOrderDao;
import com.axdoc.walletbiz.record.dao.IOrderRefundDao;
import com.axdoc.walletbiz.record.entity.PayOrder;
import com.axdoc.walletbiz.record.entity.PayOrderChart;
import com.axdoc.walletbiz.record.entity.PayOrderRefund;
import com.axdoc.walletbiz.record.exceptions.WalletBizRecordException;
import com.axdoc.walletbiz.record.service.DataService;
import com.axdoc.walletbiz.record.service.RecordService;

/**
 * 处理数据
 * 
 * @author sunzhaoli
 *
 */
@Service("dataService")
@Path("data")
@Produces({ "application/json; charset=UTF-8", "text/xml; charset=UTF-8" })
public class DataServiceImpl implements DataService {

	public static Logger logger = LoggerFactory.getLogger("data");

	// 配置参数
	private static int corePoolSize = 50;
	private static int maxPoolSize = 100;
	private static final int maxQueueSize = 2000;
	private static final int keepAliveTime = 30; // 秒

	private static int batchCount = 500;

	@Autowired
	private IOrderDao orderDao;
	@Autowired
	private IOrderRefundDao refundDao;
	@Autowired
	private RecordService recordService;
	@Autowired
	private AmqpTemplate amqpTemplate_pay;

	/**
	 * 追历史数据（患者历史流水，时间点之前的）
	 * 
	 * @param bt
	 *            处理这个时间点之前的所有数据
	 * @return
	 * @throws WalletBizRecordException
	 * @throws Exception
	 */
	public String dataBeforeBack(String bt) throws WalletBizRecordException, Exception {
		Long stl = System.currentTimeMillis();
		logger.info("dataBefore start...");
		logger.info("dataBefore param bt={}", bt);
		if (StringUtils.isBlank(bt)) {
			throw new WalletBizRecordException(WalletBizRecordException.MISSING_ARGS, "缺少参数对象bt");
		}
		SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMddHHmmss");
		SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
		Date date = sdf.parse(bt);
		bt = df.format(date);
		// 支付完成和已退款总数
		int sumCount = orderDao.queryCount(bt);
		logger.info("dataBefore sumCount={}", sumCount);
		if (sumCount < 1) {
			return "sumCount小于1,没有历史数据可处理";
		}
		int remainder = 0;// 余数
		int batchs = 0; // 多少个批次
		int offset = 0;

		// 算分多少个批次
		if (sumCount > batchCount) {
			remainder = sumCount % batchCount;
			if (remainder == 0) {
				batchs = sumCount / batchCount;
			} else {
				batchs = sumCount / batchCount + 1;
			}
		}
		if (sumCount == batchCount || sumCount < batchCount) {
			batchs = 1;
		}

		List<PayOrder> list = new ArrayList<PayOrder>();
		logger.info("dataBefore batchs={}", batchs);
		// 开始分批处理
		for (int i = 0; i < batchs; i++) {
			logger.info("dataBefore batchs now={}", (i + 1));
			if (batchs != 1) {
				offset = batchCount * i;
			}
			list = orderDao.queryPayOrderByBatch(bt, batchCount, offset);
			BeforeRunnable runnable = new BeforeRunnable(list);
			beforeThreadPool.execute(runnable);
		}

		Long etl = System.currentTimeMillis();
		logger.info("dataBefore end time={}", (etl - stl));
		return "success";
	}

	/**
	 * 追历史数据（患者历史流水，时间点之后的）
	 * 
	 * @param at
	 * @return
	 * @throws WalletBizRecordException
	 * @throws Exception
	 */
	public String dataAfterBack(String at) throws WalletBizRecordException, Exception {
		Long stl = System.currentTimeMillis();
		logger.info("dataAfter start...");
		logger.info("dataAfter param at={}", at);
		if (StringUtils.isBlank(at)) {
			throw new WalletBizRecordException(WalletBizRecordException.MISSING_ARGS, "缺少参数对象bt");
		}
		SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMddHHmmss");
		SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
		Date date = sdf.parse(at);
		at = df.format(date);
		// 支付完成和已退款总数
		int sumCount = orderDao.queryCountAfter(at);
		logger.info("dataAfter sumCount={}", sumCount);
		if (sumCount < 1) {
			return "sumCount小于1,没有历史数据可处理";
		}
		int remainder = 0;// 余数
		int batchs = 0; // 多少个批次
		int offset = 0;

		// 算分多少个批次
		if (sumCount > batchCount) {
			remainder = sumCount % batchCount;
			if (remainder == 0) {
				batchs = sumCount / batchCount;
			} else {
				batchs = sumCount / batchCount + 1;
			}
		}
		if (sumCount == batchCount || sumCount < batchCount) {
			batchs = 1;
		}

		List<PayOrder> list = new ArrayList<PayOrder>();
		logger.info("dataAfter batchs={}", batchs);
		// 开始分批处理
		for (int i = 0; i < batchs; i++) {
			logger.info("dataAfter batchs now={}", (i + 1));
			if (batchs != 1) {
				offset = batchCount * i;
			}
			list = orderDao.queryPayOrderByBatchAfter(at, batchCount, offset);
			AfterRunnable runnable = new AfterRunnable(list);
			afterThreadPool.execute(runnable);
		}

		Long etl = System.currentTimeMillis();
		logger.info("dataAfter end time={}", (etl - stl));
		return "success";
	}

	/**
	 * 追历史数据（患者历史流水，时间点之前的）
	 * 
	 * @param bt
	 *            处理这个时间点之前的所有数据
	 * @return
	 * @throws WalletBizRecordException
	 * @throws Exception
	 */
	@GET
	@POST
	@Path("/before/")
	@Consumes({ MediaType.APPLICATION_JSON })
	public String dataBefore(@QueryParam("bt") String bt) throws WalletBizRecordException, Exception {
		DataAsync async = new DataAsync(1, bt);
		async.doData();
		return "已将任务添加到后台执行!";
	}

	/**
	 * 追历史数据（患者历史流水，时间点之后的）
	 * 
	 * @param at
	 * @return
	 * @throws WalletBizRecordException
	 * @throws Exception
	 */
	@GET
	@POST
	@Path("/after/")
	@Consumes({ MediaType.APPLICATION_JSON })
	public String dataAfter(@QueryParam("at") String at) throws WalletBizRecordException, Exception {
		DataAsync async = new DataAsync(2, at);
		async.doData();
		return "已将任务添加到后台执行!";
	}

	// 采用有界阻塞队列(处理历史数据队列)
	private static final ThreadPoolExecutor beforeThreadPool = new ThreadPoolExecutor(corePoolSize, maxPoolSize,
			keepAliveTime, TimeUnit.MINUTES, new ArrayBlockingQueue<Runnable>(maxQueueSize),
			new RejectedExecutionHandler() {
				public void rejectedExecution(Runnable arg0, ThreadPoolExecutor arg1) {
					// TODO ...
				}
			});

	// 采用有界阻塞队列(处理历史数据队列)
	private static final ThreadPoolExecutor afterThreadPool = new ThreadPoolExecutor(corePoolSize, maxPoolSize,
			keepAliveTime, TimeUnit.MINUTES, new ArrayBlockingQueue<Runnable>(maxQueueSize),
			new RejectedExecutionHandler() {
				public void rejectedExecution(Runnable arg0, ThreadPoolExecutor arg1) {
					// TODO ...
				}
			});

	// 处理历史数据
	private class BeforeRunnable implements Runnable {
		List<PayOrder> list = new ArrayList<PayOrder>();

		@Override
		public void run() {
			Long stl = System.currentTimeMillis();
			Long sctl = 0L;
			Long ectl = 0L;
			int batchDataSize = 0;
			PayOrder order = new PayOrder();
			String trade_no = "";
			Long time = 0l;
			int status = 0;
			batchDataSize = list.size();
			List<PayOrderChart> poc = new ArrayList<PayOrderChart>();
			List<PayOrderRefund> por = new ArrayList<PayOrderRefund>();
			for (int j = 0; j < batchDataSize; j++) {
				sctl = System.currentTimeMillis();
				order = list.get(j);
				trade_no = order.getTrade_no();
				if (StringUtils.isBlank(trade_no)) {
					continue;
				}
				status = order.getStatus();
				if (status == 2) { // 支付完成
					poc = orderDao.queryPayOrderChart(trade_no);
					if (null == poc || poc.size() < 1) {
						continue;
					}
					for (int n = 0; n < poc.size(); n++) {
						trade_no = poc.get(n).getTrade_no();
						time = poc.get(n).getCreate_time().getTime();
						recordService.paySuccess(trade_no, time, "999999999");
					}

				}
				if (status == 5) { // 退款完成
					por = refundDao.queryByTradeNo(trade_no);
					if (null == por || por.size() < 1) {
						continue;
					}
					for (int n = 0; n < por.size(); n++) {
						trade_no = por.get(n).getTrade_no();
						time = por.get(n).getCreate_time().getTime();
						recordService.refundSuccess(trade_no, time, "999999999");
					}
				}
				ectl = System.currentTimeMillis();
				logger.info("BeforeRunnable child end time={}", (ectl - sctl));
			}
			Long etl = System.currentTimeMillis();
			logger.info("BeforeRunnable end time={}", (etl - stl));
		}

		public BeforeRunnable(List<PayOrder> list) {
			super();
			this.list = list;
		}
	}

	// 处理历史数据
	private class AfterRunnable implements Runnable {
		List<PayOrder> list = new ArrayList<PayOrder>();

		@Override
		public void run() {
			Long stl = System.currentTimeMillis();
			Long sctl = 0L;
			Long ectl = 0L;
			int batchDataSize = 0;
			PayOrder order = new PayOrder();
			String trade_no = "";
			Long time = 0l;
			int status = 0;
//			String bizTime="";
			String biz_no = "";
			String trx_no = "";
			batchDataSize = list.size();
//			SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMddHHmmss");
			List<PayOrderChart> poc = new ArrayList<PayOrderChart>();
			List<PayOrderRefund> por = new ArrayList<PayOrderRefund>();
			for (int j = 0; j < batchDataSize; j++) {
				sctl = System.currentTimeMillis();
				order = list.get(j);
				trade_no = order.getTrade_no();
				biz_no = order.getBizid();
				trx_no = order.getTransaction_id();
				if (StringUtils.isBlank(trade_no)) {
					continue;
				}
				status = order.getStatus();
				if (status == 2) { // 支付完成
					poc = orderDao.queryPayOrderChart(trade_no);
					if (null == poc || poc.size() < 1) {
						continue;
					}
					for (int n = 0; n < poc.size(); n++) {
						trade_no = poc.get(n).getTrade_no();
//						try {
//							bizTime = sdf.parse(poc.get(n).getCreate_time().toString()).toString();
//						} catch (ParseException e) {
//							logger.error("AfterRunnable error=",e);
//						}
						time = poc.get(n).getCreate_time().getTime();
						sendMQ(biz_no, trx_no, trade_no,time, 2);
					}

				}
				if (status == 5) { // 退款完成
					por = refundDao.queryByTradeNo(trade_no);
					if (null == por || por.size() < 1) {
						continue;
					}
					for (int n = 0; n < por.size(); n++) {
						trade_no = por.get(n).getTrade_no();
						time = por.get(n).getCreate_time().getTime();
						sendMQ(biz_no, trx_no, trade_no, time, 5);
					}
				}
				ectl = System.currentTimeMillis();
				logger.info("AfterRunnable child end time={}", (ectl - sctl));
			}
			Long etl = System.currentTimeMillis();
			logger.info("AfterRunnable end time={}", (etl - stl));
		}

		public AfterRunnable(List<PayOrder> list) {
			super();
			this.list = list;
		}
	}

	public void sendMQ(String biz_no, String trx_no, String trade_no, Long biz_time, int type) {
		try {
			JSONObject require = new JSONObject();
			Long occurTime = System.currentTimeMillis();
			UUID uuid = UUID.randomUUID();
			String id = uuid.toString();
			require.put("occurTime", occurTime);
			require.put("id", id);
			require.put("priority", 5);
			require.put("from", 100);
			JSONObject data = new JSONObject();
			if (StringUtils.isNotBlank(biz_no)) {
				data.put("biz_no", biz_no);
			} else {
				data.put("biz_no", "");
			}
			if (StringUtils.isNotBlank(trx_no)) {
				data.put("trx_no", trx_no);
			} else {
				data.put("trx_no", "");
			}
			data.put("trade_no", trade_no);
			data.put("biz_time", biz_time);
			if (type == 2) {
				data.put("msgType", "1501");
				require.put("type", 1501);
			}
			if (type == 5) {
				data.put("msgType", "1502");
				require.put("type", 1502);
			}
			require.put("data", data);
			if (type == 2) {
				amqpTemplate_pay.convertAndSend("1501", require.toJSONString());
			}
			if (type == 5) {
				amqpTemplate_pay.convertAndSend("1502", require.toJSONString());
			}
			

		} catch (Exception e) {
			logger.error("AfterRunnable error=", e);
		}
	}

}
